package com.dmp.storm.topology;

import backtype.storm.Config;
import backtype.storm.generated.AlreadyAliveException;
import backtype.storm.generated.AuthorizationException;
import backtype.storm.generated.InvalidTopologyException;
import com.dmp.util.Constant;

import java.io.File;

/**
 * Created by wangqiang on 2016/4/14.
 */
public class XmlParser {

    private ITopologyCtx ctx;
    private String topologyName;  //topology名称
    private String workNum; //work节点数量
    private String xmlName; //xml名称
    private String bootMode; //启动模式

    /**
     * @param topologyName topology名称
     * @param xmlName      xml名称
     * @param args     参数
     */
    public XmlParser(String topologyName,  String xmlName, String[] args) {
        String workNum = "";
        String bootMode = "";
        if (args.length > 0) {
            bootMode = args[0];
            if (args.length > 1) {
                workNum = args[1];
            }
        }
    	
        this.topologyName = topologyName;
        this.workNum = workNum;
        this.xmlName = xmlName;
        this.bootMode = bootMode;

        File file = new File("../etc/" + xmlName);
        if (file.exists()) {
            ctx = new XmlTopologyCtx(XmlTopologyCtx.SS.CUSTOM, file.getAbsolutePath());
        } else {
            ctx = new XmlTopologyCtx(XmlTopologyCtx.SS.LOCAL, xmlName);
        }
    }

    public void run() {
        try {
            if (("serv").equals(bootMode)) {
                Config conf = new Config();
                int numWorkers = Constant.StormDef.STORM_NUM_WORKERS_DEFAULT;
                if (!"".equals(workNum)) {
                    numWorkers = Long.valueOf(workNum).intValue();
                }
                conf.setNumWorkers(numWorkers);
                conf.put(Config.TOPOLOGY_ENABLE_MESSAGE_TIMEOUTS, false);
                conf.put(Config.TOPOLOGY_MESSAGE_TIMEOUT_SECS, 10000);
                ctx.start(topologyName, conf);

            } else {
                ctx.startLocal();
            }
        } catch (InvalidTopologyException e) {
            e.printStackTrace();
        } catch (AlreadyAliveException e) {
            e.printStackTrace();
        } catch (AuthorizationException e) {
            e.printStackTrace();
        }
    }

}
